package com.example.reactor.reactor;

import java.util.Arrays;
import java.util.function.Function;

public class ReactorExample {
    public static void main(String[] args) {
        Mono<Integer> mono = new Mono<>(42);
        Flux<String> flux = new Flux<>(Arrays.asList("A", "B", "C"));

        // 测试 map 操作符
        mono.map(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer value) {
                System.out.println(value);
                return value * 2;
            }
        }).subscribe(new MySubscriber<>(
                value -> System.out.println("Mono map received: " + value),
                () -> System.out.println("Mono completed."),
                error -> System.out.println("Mono error: " + error)
        ));
        System.out.println("---------------------------------------分割线-------------------------------------");


        // 测试 flatMap 操作符
        flux.flatMap(value -> new Flux<>(Arrays.asList(value + "1", value + "2")))
                .subscribe(new MySubscriber<>(
                        value -> System.out.println("Flux flatMap received: " + value),
                        () -> System.out.println("Flux completed."),
                        error -> System.out.println("Flux error: " + error)
                ));
        System.out.println("---------------------------------------分割线-------------------------------------");

        // 测试 filter 操作符
        flux.filter(value -> value.equals("B"))
                .subscribe(new MySubscriber<>(
                        value -> System.out.println("Flux filter received: " + value),
                        () -> System.out.println("Flux completed."),
                        error -> System.out.println("Flux error: " + error)
                ));
        System.out.println("---------------------------------------分割线-------------------------------------");

        // 测试 merge 操作符
        Flux<String> flux2 = new Flux<>(Arrays.asList("D", "E"));
        Flux.merge(flux, flux2).subscribe(new MySubscriber<>(
                value -> System.out.println("Flux merge received: " + value),
                () -> System.out.println("Flux completed."),
                error -> System.out.println("Flux error: " + error)
        ));
        System.out.println("---------------------------------------分割线-------------------------------------");

        // 测试 concat 操作符
        Flux.concat(flux, flux2).subscribe(new MySubscriber<>(
                value -> System.out.println("Flux concat received: " + value),
                () -> System.out.println("Flux completed."),
                error -> System.out.println("Flux error: " + error)
        ));
        System.out.println("---------------------------------------分割线-------------------------------------");

        // 测试 zip 操作符
        Flux<String> flux3 = new Flux<>(Arrays.asList("1", "2", "3"));
        Flux.zip(flux, flux3).subscribe(new MySubscriber<>(
                value -> System.out.println("Flux zip received: " + value.getKey() + " " + value.getValue()),
                () -> System.out.println("Flux completed."),
                error -> System.out.println("Flux error: " + error)
        ));
        System.out.println("---------------------------------------分割线-------------------------------------");

        // 测试 buffer 操作符
        flux.buffer(2).subscribe(new MySubscriber<>(
                value -> System.out.println("Flux buffer received: " + value),
                () -> System.out.println("Flux completed."),
                error -> System.out.println("Flux error: " + error)
        ));
        System.out.println("---------------------------------------分割线-------------------------------------");

        // 测试 doOnNext 操作符
        flux.doOnNext(value -> System.out.println("Flux doOnNext received: " + value))
                .subscribe(new MySubscriber<>(value -> {
                }, () -> {
                }, error -> {
                }));
        System.out.println("---------------------------------------分割线-------------------------------------");

        // 测试 doOnComplete 操作符
        mono.doOnComplete(() -> System.out.println("Mono completed"))
                .subscribe(new MySubscriber<>(value -> {
                }, () -> {
                }, error -> {
                }));
        System.out.println("---------------------------------------分割线-------------------------------------");

        // 测试 doOnError 操作符
        flux.doOnError(error -> System.out.println("Flux error: " + error))
                .subscribe(new MySubscriber<>(value -> {
                }, () -> {
                }, error -> System.out.println("Handled error")));
        System.out.println("---------------------------------------分割线-------------------------------------");

        // 测试异步执行
        Schedulers.executeAsync(() -> {
            System.out.println("Running asynchronously");
            mono.subscribe(new MySubscriber<>(value -> System.out.println("Async Mono received: " + value), () -> {
            }, error -> {
            }));
        });
        System.out.println("---------------------------------------分割线-------------------------------------");

        // 停止 Reactor
        Schedulers.shutdown();
    }
}